package com.eco.fishway.listener;

import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.eco.db.entity.Scan;
import com.eco.db.entity.Sensor;
import com.eco.fishway.service.ScanService;
import com.eco.fishway.service.SensorService;
import com.github.tocrhz.mqtt.annotation.MqttSubscribe;
import com.github.tocrhz.mqtt.annotation.Payload;
import com.influxdb.client.WriteApi;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Author : xulinyu
 * @Date: 2024/7/2 16:04
 * @Description:
 */
@Component
@Slf4j
public class MqttMessageListener {


    private final SensorService sensorService;

    private final ScanService scanService;

    private final WriteApi writeApi;

    public MqttMessageListener(WriteApi writeApi, SensorService sensorService, ScanService scanService) {
        this.writeApi = writeApi;
        this.sensorService = sensorService;
        this.scanService = scanService;
    }

    /**
     * {
     *   "devList": [
     *     {
     *       "varList": {
     *         "dissolved_oxygen_in_freshwater_pools": 3.66,
     *         "dissolved_oxygen_temperature_in_freshwater_pool": 27.1
     *       },
     *       "devSn": "zt_freshwater_dissolved_oxygen",
     *       "devSort": "",
     *       "ts": 1721270640
     *     }
     *   ],
     *   "cmdId": 103,
     *   "type": 0,
     *   "ver": "0.5.1.0",
     *   "seq": "0e285865-d82e-4c71-85f2-3f183bc60faa",
     *   "time": "2024-07-18 10:44:00"
     * }
     */

    /**
     * topic = test/+
     */
    @MqttSubscribe("/sys/WG583LQ2S24071702942/up")
    public void sub(String topic, MqttMessage message, @Payload String payload) {
        log.info("receive from    : {}", topic);
        log.info("message payload : {}", new String(message.getPayload(), StandardCharsets.UTF_8));
        //payload有些键值是重复的
        JSONObject payloadObj = JSONUtil.parseObj(payload);
        String cmdId = payloadObj.getStr("cmdId");
        String timeStr = payloadObj.getStr("time");
        //1720494780000 这个时间精度是ms
        if (StringUtils.isNotEmpty(timeStr)) {
            if ("103".equals(cmdId)) {
                JSONArray devList = payloadObj.getJSONArray("devList");
                List<JSONObject> list = JSONUtil.toList(devList, JSONObject.class);
                list.forEach(x -> {
                    JSONObject varList = x.getJSONObject("varList");
                    log.info("=============开始推送mqtt数据=============");
                    //写入 mysql begin
                    Double lsy1 = varList.getDouble("ls1");
                    if(lsy1 != null) {
                        Double lsy2 = varList.getDouble("ls2");
                        Double lsy3 = varList.getDouble("ls3");
                        Double lsy4 = varList.getDouble("ls4");
                        Double lsy5 = varList.getDouble("ls5");
                        Double lsy6 = varList.getDouble("ls6");
                        Sensor sensor = new Sensor();
                        sensor.setLsy1(lsy1);
                        sensor.setLsy2(lsy2);
                        sensor.setLsy3(lsy3);
                        sensor.setLsy4(lsy4);
                        sensor.setLsy5(lsy5);
                        sensor.setLsy6(lsy6);
                        //utc时间转东八区时间
                        sensor.setStime(DateUtil.parse(timeStr));
                        sensorService.save(sensor);
                        log.info("save sensor : {}", JSONUtil.toJsonStr(sensor));
                    }
                    Double fs = varList.getDouble("fs");
                    if(fs != null) {
                        Double fx = varList.getDouble("fx");
                        Double dqy = varList.getDouble("dqy");
                        Double qw = varList.getDouble("qw");
                        Double sd = varList.getDouble("sd");
                        Double co2 = varList.getDouble("CO2");
                        Double gzqd = varList.getDouble("gzqd");
                        Double yl = varList.getDouble("yl");
                        Double sw = varList.getDouble("sw");
                        Double ph = varList.getDouble("ph");
                        Double rjy = varList.getDouble("rjy");
                        Double ad = varList.getDouble("ad");
                        Double zd = varList.getDouble("zd");
                        Scan scan = new Scan();
                        scan.setStime(DateUtil.parse(timeStr));
                        scan.setWfs(fs);
                        scan.setWfx(fx);
                        scan.setWqy(dqy);
                        scan.setWwd(qw);
                        scan.setWsd(sd);
                        scan.setWco2(co2);
                        scan.setWgz(gzqd);
                        scan.setWyl(yl);
                        scan.setSzwd(sw);
                        scan.setSzph(ph);
                        scan.setSzry(rjy);
                        scan.setSznh4(ad);
                        scan.setSzzd(zd);
                        scanService.save(scan);
                        log.info("save scan : {}", JSONUtil.toJsonStr(scan));
                    }

                });
                //写入 mysql end
            }
        }
    }
}